本文解释一下为什么以及如何使用一个 ProtocolCodecFilter。
你的应用接收到的基本上是一串字节,你需要将这些字节转换为消息 (高层对象)。
有三种常见的技术用来将字节流分割为消息:
本文将使用第一种和第二种方法,因为它们很明显更容易实现。然后我们再看一下使用定界符的做法。
我们将会开发一个 (毫无用处的) 图形字符发生器协议服务器来说明如何实现你自己协议的编解码器 (ProtocolEncoder、ProtocolDecoder 和 ProtocolCodecFactory)。这个协议是很简单的。这是请求消息的布局:
4 bytes | 4 bytes | 4 bytes |
---|---|---|
width | height | numchars |
服务器以符合请求尺寸的两张图片进行响应,每张图片上打上请求的字符数。这是一个响应消息的布局:
4 bytes | variable length body | 4 bytes | variable length body |
---|---|---|---|
length1 | image1 | length2 | image2 |
我们需要的用于请求和响应的编码和解码的类的概述:
ImageRequest 类源代码如下:
public class ImageRequest {
private int width;
private int height;
private int numberOfCharacters;
public ImageRequest(int width, int height, int numberOfCharacters) {
this.width = width;
this.height = height;
this.numberOfCharacters = numberOfCharacters;
}
public int getWidth() {
return width;
}
public int getHeight() {
return height;
}
public int getNumberOfCharacters() {
return numberOfCharacters;
}
}
编码往往比解码容易,因此我们先从 ImageRequestEncoder 开始:
public class ImageRequestEncoder implements ProtocolEncoder {
public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception {
ImageRequest request = (ImageRequest) message;
IoBuffer buffer = IoBuffer.allocate(12, false);
buffer.putInt(request.getWidth());
buffer.putInt(request.getHeight());
buffer.putInt(request.getNumberOfCharacters());
buffer.flip();
out.write(buffer);
}
public void dispose(IoSession session) throws Exception {
// nothing to dispose
}
}
备注:
现在我们来看一下解码器。CumulativeProtocolDecoder 绝对对写你自己的编码器有很大帮助:它将把你的解码器决定对连入数据可以做一些事情之前都缓存起来。在这种情况下消息具有固定大小,因此很容易等待所有的数据到齐以后再进行一些操作:
public class ImageRequestDecoder extends CumulativeProtocolDecoder {
protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
if (in.remaining() >= 12) {
int width = in.getInt();
int height = in.getInt();
int numberOfCharachters = in.getInt();
ImageRequest request = new ImageRequest(width, height, numberOfCharachters);
out.write(request);
return true;
} else {
return false;
}
}
}
备注:
响应也是一个非常简单的 POJO:
public class ImageResponse {
private BufferedImage image1;
private BufferedImage image2;
public ImageResponse(BufferedImage image1, BufferedImage image2) {
this.image1 = image1;
this.image2 = image2;
}
public BufferedImage getImage1() {
return image1;
}
public BufferedImage getImage2() {
return image2;
}
}
响应的编码也很简单:
public class ImageResponseEncoder extends ProtocolEncoderAdapter {
public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception {
ImageResponse imageResponse = (ImageResponse) message;
byte[] bytes1 = getBytes(imageResponse.getImage1());
byte[] bytes2 = getBytes(imageResponse.getImage2());
int capacity = bytes1.length + bytes2.length + 8;
IoBuffer buffer = IoBuffer.allocate(capacity, false);
buffer.setAutoExpand(true);
buffer.putInt(bytes1.length);
buffer.put(bytes1);
buffer.putInt(bytes2.length);
buffer.put(bytes2);
buffer.flip();
out.write(buffer);
}
private byte[] getBytes(BufferedImage image) throws IOException {
ByteArrayOutputStream baos = new ByteArrayOutputStream();
ImageIO.write(image, "PNG", baos);
return baos.toByteArray();
}
}
备注:
当无法提前预测 IoBuffer 的长度时,你可以通过调用 buffer.setAutoExpand(true); 使用一个自动扩展缓存
现在我们来看一下响应的解码:
public class ImageResponseDecoder extends CumulativeProtocolDecoder {
private static final String DECODER_STATE_KEY = ImageResponseDecoder.class.getName() + ".STATE";
public static final int MAX_IMAGE_SIZE = 5 * 1024 * 1024;
private static class DecoderState {
BufferedImage image1;
}
protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
DecoderState decoderState = (DecoderState) session.getAttribute(DECODER_STATE_KEY);
if (decoderState == null) {
decoderState = new DecoderState();
session.setAttribute(DECODER_STATE_KEY, decoderState);
}
if (decoderState.image1 == null) {
// try to read first image
if (in.prefixedDataAvailable(4, MAX_IMAGE_SIZE)) {
decoderState.image1 = readImage(in);
} else {
// not enough data available to read first image
return false;
}
}
if (decoderState.image1 != null) {
// try to read second image
if (in.prefixedDataAvailable(4, MAX_IMAGE_SIZE)) {
BufferedImage image2 = readImage(in);
ImageResponse imageResponse = new ImageResponse(decoderState.image1, image2);
out.write(imageResponse);
decoderState.image1 = null;
return true;
} else {
// not enough data available to read second image
return false;
}
}
return false;
}
private BufferedImage readImage(IoBuffer in) throws IOException {
int length = in.getInt();
byte[] bytes = new byte[length];
in.get(bytes);
ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
return ImageIO.read(bais);
}
}
备注:
如果响应只有单一的一个图片,我们就无需保存解码状态了:
protected boolean doDecode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws Exception {
if (in.prefixedDataAvailable(4)) {
int length = in.getInt();
byte[] bytes = new byte[length];
in.get(bytes);
ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
BufferedImage image = ImageIO.read(bais);
out.write(image);
return true;
} else {
return false;
}
}
现在我们把它们都组合在一起:
public class ImageCodecFactory implements ProtocolCodecFactory {
private ProtocolEncoder encoder;
private ProtocolDecoder decoder;
public ImageCodecFactory(boolean client) {
if (client) {
encoder = new ImageRequestEncoder();
decoder = new ImageResponseDecoder();
} else {
encoder = new ImageResponseEncoder();
decoder = new ImageRequestDecoder();
}
}
public ProtocolEncoder getEncoder(IoSession ioSession) throws Exception {
return encoder;
}
public ProtocolDecoder getDecoder(IoSession ioSession) throws Exception {
return decoder;
}
}
备注:
这里是服务器端对 ProtocolCodecFactory 的使用:
public class ImageServer {
public static final int PORT = 33789;
public static void main(String[] args) throws IOException {
ImageServerIoHandler handler = new ImageServerIoHandler();
NioSocketAcceptor acceptor = new NioSocketAcceptor();
acceptor.getFilterChain().addLast("protocol", new ProtocolCodecFilter(new ImageCodecFactory(false)));
acceptor.setLocalAddress(new InetSocketAddress(PORT));
acceptor.setHandler(handler);
acceptor.bind();
System.out.println("server is listenig at port " + PORT);
}
}
客户端的使用完全一致:
public class ImageClient extends IoHandlerAdapter {
public static final int CONNECT_TIMEOUT = 3000;
private String host;
private int port;
private SocketConnector connector;
private IoSession session;
private ImageListener imageListener;
public ImageClient(String host, int port, ImageListener imageListener) {
this.host = host;
this.port = port;
this.imageListener = imageListener;
connector = new NioSocketConnector();
connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(new ImageCodecFactory(true)));
connector.setHandler(this);
}
public void messageReceived(IoSession session, Object message) throws Exception {
ImageResponse response = (ImageResponse) message;
imageListener.onImages(response.getImage1(), response.getImage2());
}
...
完整性考虑,现在附加一个服务器端的 IoHandler 代码:
public class ImageServerIoHandler extends IoHandlerAdapter {
private final static String characters = "mina rocks abcdefghijklmnopqrstuvwxyz0123456789";
public static final String INDEX_KEY = ImageServerIoHandler.class.getName() + ".INDEX";
private Logger logger = LoggerFactory.getLogger(this.getClass());
public void sessionOpened(IoSession session) throws Exception {
session.setAttribute(INDEX_KEY, 0);
}
public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
IoSessionLogger sessionLogger = IoSessionLogger.getLogger(session, logger);
sessionLogger.warn(cause.getMessage(), cause);
}
public void messageReceived(IoSession session, Object message) throws Exception {
ImageRequest request = (ImageRequest) message;
String text1 = generateString(session, request.getNumberOfCharacters());
String text2 = generateString(session, request.getNumberOfCharacters());
BufferedImage image1 = createImage(request, text1);
BufferedImage image2 = createImage(request, text2);
ImageResponse response = new ImageResponse(image1, image2);
session.write(response);
}
private BufferedImage createImage(ImageRequest request, String text) {
BufferedImage image = new BufferedImage(request.getWidth(), request.getHeight(), BufferedImage.TYPE_BYTE_INDEXED);
Graphics graphics = image.createGraphics();
graphics.setColor(Color.YELLOW);
graphics.fillRect(0, 0, image.getWidth(), image.getHeight());
Font serif = new Font("serif", Font.PLAIN, 30);
graphics.setFont(serif);
graphics.setColor(Color.BLUE);
graphics.drawString(text, 10, 50);
return image;
}
private String generateString(IoSession session, int length) {
Integer index = (Integer) session.getAttribute(INDEX_KEY);
StringBuffer buffer = new StringBuffer(length);
while (buffer.length() < length) {
buffer.append(characters.charAt(index));
index++;
if (index >= characters.length()) {
index = 0;
}
}
session.setAttribute(INDEX_KEY, index);
return buffer.toString();
}
}
关于编码和解码不仅于此。但是我希望本文足以让你开始了。不久的将来我会尝试着添加关于 DemuxingProtocolCodecFactory 的一些介绍。届时我们也将看一下如何使用定界符取代长度前缀的做法
译者注:翻译版本的项目源码见 https://github.com/waylau/apache-mina-2-user-guide-demos 中的com.waylau.mina.demo.imagine
包下